package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.VisitorStats;
import com.atguigu.realtime.util.MySinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/4/24 14:18
 */
public class DWSVisitorStatsApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DWSVisitorStatsApp().init(4001, 2, "DWSVisitorStatsApp", "DWSVisitorStatsApp_1",
                                      "dwd_page", "dwm_uv", "dwm_user_jump_detail");
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStreamSource<String>> streams) {
        // 1. 把多个流的数据, 合并成一个流 union
        DataStream<VisitorStats> visitorStatsStream = parseStreamAndUnionOne(streams);
        visitorStatsStream.print("visitorStatsStream");
        
        // 2. 安装窗口和各个维度进行聚合
        SingleOutputStreamOperator<VisitorStats> statsStream = aggregateByWindowAndDim(visitorStatsStream);
        
        // 3. 写出数据到clickhouse中
        sendToClickhouse(statsStream);
    
    }
    
    private void sendToClickhouse(SingleOutputStreamOperator<VisitorStats> statsStream) {
        statsStream.print();
        statsStream.addSink(MySinkUtil.getClickHouseSink("gmall2021", "visitor_stats_2021", VisitorStats.class));
    
    }
    
    private SingleOutputStreamOperator<VisitorStats> aggregateByWindowAndDim(DataStream<VisitorStats> visitorStatsStream) {
        return visitorStatsStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                    .withTimestampAssigner((vs, ts) -> vs.getTs())
            )
            .keyBy(vs -> vs.getVc() + "_" + vs.getCh() + "_" + vs.getAr() + "_" + vs.getIs_new())
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats value1,
                                               VisitorStats value2) throws Exception {
                        
                        value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                        value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                        value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                        value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                        value1.setDur_sum(value1.getDur_sum() + value2.getDur_sum());
                        return value1;
                    }
                },
                new ProcessWindowFunction<VisitorStats, VisitorStats, String, TimeWindow>() {
                    
                    private SimpleDateFormat sdf;
                    
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }
                    
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<VisitorStats> elements,
                                        Collector<VisitorStats> out) throws Exception {
                        
                        VisitorStats vs = elements.iterator().next();
                        
                        TimeWindow window = ctx.window();
                        vs.setStt(sdf.format(window.getStart()));
                        vs.setEdt(sdf.format(window.getEnd()));
                        vs.setTs(window.getEnd());
                        
                        out.collect(vs);
                        
                    }
                }
            );
        
    }
    
    private DataStream<VisitorStats> parseStreamAndUnionOne(Map<String, DataStreamSource<String>> streams) {
        DataStreamSource<String> pageStream = streams.get("dwd_page");
        DataStreamSource<String> uvStream = streams.get("dwm_uv");
        DataStreamSource<String> userJumpStream = streams.get("dwm_user_jump_detail");
        
        // 1. 计算pv  // 5. 持续访问时间
        SingleOutputStreamOperator<VisitorStats> pageStatsStream = pageStream
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                JSONObject page = obj.getJSONObject("page");
                return new VisitorStats("", "",
                                        common.getString("vc"),
                                        common.getString("ch"),
                                        common.getString("ar"),
                                        common.getString("is_new"),
                                        0L, 1L, 0L, 0L, page.getLong("during_time"),
                                        obj.getLong("ts"));
            });
        
        // 2. 计算uv
        SingleOutputStreamOperator<VisitorStats> uvStatsStream = uvStream
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                return new VisitorStats("", "",
                                        common.getString("vc"),
                                        common.getString("ch"),
                                        common.getString("ar"),
                                        common.getString("is_new"),
                                        1L, 0L, 0L, 0L, 0L,
                                        obj.getLong("ts"));
            });
        
        // 3. 计算进入次数
        SingleOutputStreamOperator<VisitorStats> svStatsStream = pageStream
            .flatMap(new FlatMapFunction<String, VisitorStats>() {
                @Override
                public void flatMap(String json,
                                    Collector<VisitorStats> out) throws Exception {
                    JSONObject obj = JSON.parseObject(json);
                    JSONObject common = obj.getJSONObject("common");
                    String lastPageId = obj.getJSONObject("page").getString("last_page_id");
                    if (lastPageId == null || lastPageId.length() == 0) {
                        VisitorStats visitorStats = new VisitorStats("", "",
                                                                     common.getString("vc"),
                                                                     common.getString("ch"),
                                                                     common.getString("ar"),
                                                                     common.getString("is_new"),
                                                                     0L, 0L, 1L, 0L, 0L,
                                                                     obj.getLong("ts"));
                        out.collect(visitorStats);
                    }
                }
            });
        // 4. 跳出次数
        SingleOutputStreamOperator<VisitorStats> ujStatsStream = userJumpStream
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                return new VisitorStats("", "",
                                        common.getString("vc"),
                                        common.getString("ch"),
                                        common.getString("ar"),
                                        common.getString("is_new"),
                                        0L, 0L, 0L, 1L, 0L,
                                        obj.getLong("ts"));
            });
        
        // 6. 把四个流合并成一个流 union
        
        return pageStatsStream.union(uvStatsStream, ujStatsStream, svStatsStream);
        
    }
}
